package asdf.trial.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicTest
{

    public static void main(String[] args) throws Exception
    {
        final String queue_name = "test.topic";
        ConnectionFactory connFactory = new ActiveMQConnectionFactory(
                "ethankserver", "serverethank", "tcp://123.57.231.228:61616");
        Connection conn = connFactory.createConnection();
        conn.start();

        final Session session = conn.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

        Topic topic = session.createTopic(queue_name);
        session.createConsumer(topic, "aaa = TRUE")
                .setMessageListener(new TopicMessageListener("subscriberB"));
        session.createConsumer(topic, "value = 123")
                .setMessageListener(new TopicMessageListener("subscriberC"));

        final MessageProducer producer = session.createProducer(topic);
        Thread producerRunner = new Thread()
        {
            @Override
            public void run()
            {
                int n = 5;
                while ( n-- > 0 ) {
                    try {
                        TextMessage msg =
                                session.createTextMessage("@@@ " + System.currentTimeMillis());
                        msg.setBooleanProperty("aaa", true);
                        System.out.println("=====> " + msg.getText());
                        producer.send(msg);
                        Thread.sleep(500);
                    } catch ( Exception e ) {
                        e.printStackTrace();
                    }
                }
            }

        };
        producerRunner.start();

        producerRunner.join();

    }

    private static class TopicMessageListener implements MessageListener
    {
        private String name;

        public TopicMessageListener(String name)
        {
            this.name = name;
        }

        @Override
        public void onMessage(Message message)
        {
            if ( message == null ) {
                System.out.println(this.name + ": a null message received.");
            } else {
                TextMessage tm = (TextMessage) message;
                try {
                    System.out.println(this.name + " <===== " + tm.getText());
                } catch ( JMSException e ) {
                    e.printStackTrace();
                }
            }
        }

    }

}
